package com.danbay.framework.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Deserializer;
import org.eclipse.californium.core.network.serialization.DataParser;

import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @author SJG
 *         2017/11/13.
 */
@Slf4j
public class AccessPackageDeserializer<V> implements Deserializer<AccessPackage<V>> {
    private String encoding = "utf8";
    /**
     * 数据包最大长度10K
     */
    private static final int MAX_PACKAGE_SIZE = 10240;

    private ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null) {
            encodingValue = configs.get("serializer.encoding");
        }

        if (encodingValue != null && encodingValue instanceof String) {
            this.encoding = (String) encodingValue;
        }
    }

    @Override
    public AccessPackage<V> deserialize(String topic, byte[] bytes) {
        try {
            AccessPackage<V> accessPackage = new AccessPackage<>();
            //先解access kafka的包
            ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
            //4字节header长度
            int headerSize = byteBuffer.getInt();
            if (headerSize < 0) {
                log.error("error header size:{}", headerSize);
                return null;
            }
            int bytesLen = bytes.length;
            if (bytesLen > MAX_PACKAGE_SIZE || headerSize > bytesLen) {
                log.error("error package size:{},header size:{},max package size:{}", bytesLen, headerSize, MAX_PACKAGE_SIZE);
                return null;
            }
            //读取header
            byte[] headerBytes = new byte[headerSize];
            byteBuffer.get(headerBytes);
            //转成header对象
            AccessHeader header = objectMapper.readValue(new String(headerBytes, encoding), AccessHeader.class);
            //校验header
            String errorMsg = header.validate();
            if (errorMsg != null) {
                log.error(errorMsg);
                return null;
            }
            accessPackage.setHeader(header);
            //消息体
            byte[] body = new byte[byteBuffer.remaining()];
            byteBuffer.get(body);

            //暂时只支持CoAP包
            DataParser dataParser = new DataParser(body);
            if (dataParser.isRequest()) {
                accessPackage.setData((V) dataParser.parseRequest());
            } else if (dataParser.isResponse()) {
                accessPackage.setData((V) dataParser.parseResponse());
            }

            log.info("deserialize received msg:{} ", accessPackage.toString());
            return accessPackage;
        } catch (Exception e) {
            log.error("deserialize error", e);
        }
        return null;
    }

    @Override
    public void close() {

    }
}
